package org.langqiao.disruptor;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;

public class LogEventMain {
	public static void main(String[] args) throws InterruptedException {
		
		//4.定义用于事件处理的线程池
		//Disruptor 通过 java.util.concurrent.ExecutorService 提供的线程来触发 Consumer 的事件处理。例如：
		ExecutorService executor = Executors.newCachedThreadPool();
		//5.指定等待策略
		/**
		 * BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；
		   SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；
		   YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。
		 */
		WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
		WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
		WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
		
		//6.启动disruptor
		LongEventFactory eventFactory = new LongEventFactory();
		int ringBufferSize = 1024 * 1024;
		//创建disruptor
		/**
		 * 1.eventFactory:事件工厂对象，创建一个一个LongEvent，LongEvent是实际的消费数据
		 * 2.ringBufferSize:缓冲区大小，必须是2的N次方
		 * 3.executor
		 * 4.ProducerType,是SINGLE还是MULTI
		 * 5.WaitStrategy,等待策略
		 */
		Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory, ringBufferSize, executor);
		//连接消费事件的方法
		disruptor.handleEventsWith(new LongEventHandler());
		disruptor.start();
		
		//发布事件
		/**
		 * Disruptor 的事件发布过程是一个两阶段提交的过程：
　　		第一步：先从 RingBuffer 获取下一个可以写入的事件的序号；
　　		第二步：获取对应的事件对象，将数据写入事件对象；
　　		第三部：将事件提交到 RingBuffer;
		事件只有在提交之后才会通知 EventProcessor 进行处理；
		 */
		//使用该方法获取存放数据的ringBuffer(环形结构)
		RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
		//LongEventProducer producer = new LongEventProducer(ringBuffer);
		LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
		ByteBuffer bb = ByteBuffer.allocate(8); 
		for(long l=0;l<100;l++){
			bb.putLong(0, l);
			producer.onData(bb);
			//Thread.sleep(1000);
		}
		//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
		disruptor.shutdown();
		//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；
		executor.shutdown();
	}
}
